package demo.mq.rabbit;

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import demo.java.lang.ThreadDemo;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Random;
import java.util.concurrent.TimeoutException;

public class FanoutExchangeDemo {

    private static Logger logger = LoggerFactory.getLogger(FanoutExchangeDemo.class);

    @Test
    public void fanOutExchangeDemo() {
        String exchangeName = "logs";
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        /**
         * 消息持久化： 1、 设置队列 durable = true
         * 2、发送的消息指定属性MessageProperties.PERSISTENT_TEXT_PLAIN
         */
        Thread producer = new Thread(new Runnable() {
            @Override
            public void run() {
                try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {
                    channel.exchangeDeclare(exchangeName, "fanout");
                    for (int i = 0; i < 100; i++) {
                        String message = i + " 滚滚长江东逝水!";
                        /*
                         * We need to supply a routingKey when sending, but its value is ignored for
                         * "fan out" exchanges. The messages will be lost if no queue is bound to the
                         * exchange yet
                         */
                        channel.basicPublish(exchangeName, "", null, message.getBytes("UTF-8"));
                        logger.info(" [x] Sent '" + message + "'");
                        ThreadDemo.safeSleep(500);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
        }, "producer");
        Random random = new Random();
        Runnable consumerTask = new Runnable() {
            @Override
            public void run() {
                int n = random.nextInt(100);
                try {
                    Connection connection = factory.newConnection();
                    Channel channel = connection.createChannel();
                    channel.exchangeDeclare(exchangeName, "fanout");
                    /*
                     * queueDeclare() create a non-durable, exclusive, autoDelete queue with a
                     * generated name:
                     */
                    String queueName = channel.queueDeclare().getQueue();
                    // 绑定 queue 和 exchange
                    channel.queueBind(queueName, exchangeName, "");

                    // accept only one unack-ed message at a time (see below)
                    channel.basicQos(1);
                    logger.info("n +  [*] Waiting for messages. To exit press CTRL+C");

                    DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), "UTF-8");
                        logger.info(n + " [x] Received '" + message + "'");
                    };
                    // acknowledgment is covered below
                    boolean autoAck = true;
                    String consumerTag = channel.basicConsume(queueName, autoAck, deliverCallback,
                            new CancelCallback() {
                                @Override
                                public void handle(String consumerTag) throws IOException {
                                    logger.info("consumerTag : {} ", consumerTag);
                                    ThreadDemo.safeSleep(1000);
                                }
                            });
                    logger.info("consumer [{}] Waiting for messages", consumerTag);
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }

            }
        };

        Thread consumer1 = new Thread(consumerTask, "consumer1");
        Thread consumer2 = new Thread(consumerTask, "consumer2");

        producer.start();
        ThreadDemo.safeSleep(3 * 1000);
        consumer1.start();
        ThreadDemo.safeSleep(3 * 1000);
        consumer2.start();
        ThreadDemo.safeSleep(3 * 60 * 1000);
    }
}
